package etcd

import (
	"context"
	"fmt"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/yyangl/yoyo/registry"
	"go.etcd.io/etcd/clientv3"
	"time"
)

type etcdRegister struct {
	client        *clientv3.Client
	cancelFunc    func()
	lease         clientv3.Lease
	leaseResp     *clientv3.LeaseGrantResponse
	keepAliveChan <-chan *clientv3.LeaseKeepAliveResponse
	ch            chan *registry.WatchChan
}

func (e *etcdRegister) GetService(key string) (map[string]string, error) {
	values := make(map[string]string, 0)
	resp, err := e.client.Get(context.TODO(), key, clientv3.WithPrefix())
	if err != nil {
		return values, err
	}
	for i := range resp.Kvs {
		key := string(resp.Kvs[i].Key)
		value := string(resp.Kvs[i].Value)
		values[key] = value
	}
	return values, nil
}

func NewEtcdRegister(addr []string) (registry.Register, error) {
	conf := clientv3.Config{
		Endpoints:   addr,
		DialTimeout: 5 * time.Second,
	}
	client, err := clientv3.New(conf)
	if err != nil {
		return nil, err
	}
	return &etcdRegister{
		client: client,
	}, nil
}

func (e *etcdRegister) setLease(time int64) error {
	lease := clientv3.NewLease(e.client)
	//设置租约时间
	leaseResp, err := lease.Grant(context.TODO(), time)
	if err != nil {
		return err
	}

	//设置续租
	ctx, cancelFunc := context.WithCancel(context.TODO())
	leaseRespChan, err := lease.KeepAlive(ctx, leaseResp.ID)
	if err != nil {
		return err
	}
	e.cancelFunc = cancelFunc
	e.keepAliveChan = leaseRespChan
	e.lease = lease
	e.leaseResp = leaseResp
	return nil
}

func (e *etcdRegister) RevokeLease() error {
	e.cancelFunc()
	_, err := e.lease.Revoke(context.TODO(), e.leaseResp.ID)
	return err
}

//监听 续租情况
func (e *etcdRegister) ListenLeaseRespChan() {
	for {
		select {
		case leaseKeepResp := <-e.keepAliveChan:
			if leaseKeepResp == nil {
				fmt.Printf("已经关闭续租功能\n")
				return
			} else {
				fmt.Printf("续租成功\n")
			}
		}
	}
}

func (e *etcdRegister) PutService(key, val string, time int64) error {
	if err := e.setLease(time); err != nil {
		return err
	}
	go e.ListenLeaseRespChan()
	kv := clientv3.NewKV(e.client)
	_, err := kv.Put(context.TODO(), key, val, clientv3.WithLease(e.leaseResp.ID))
	return err
}

func (e *etcdRegister) Watch(key string) <-chan *registry.WatchChan {
	ch := make(chan *registry.WatchChan, registry.WatchSize)
	e.ch = ch
	go e.watch(key)
	return ch
}

func (e *etcdRegister) watch(key string) {
	defer func() {
		// TODO close chan
		//close(e.ch)
	}()
	ch := e.client.Watch(context.TODO(), key, clientv3.WithPrefix())
	for resp := range ch {
		for _, ev := range resp.Events {
			v := &registry.WatchChan{}
			switch ev.Type {
			case mvccpb.PUT:
				v.Event = registry.PUT
				v.Key = string(ev.Kv.Key)
				v.Value = string(ev.Kv.Value)
			case mvccpb.DELETE:
				v.Event = registry.DELETE
				v.Key = string(ev.Kv.Key)
			}
			e.ch <- v
		}
	}
}

//
//func (e *etcdRegister) Listen() <-chan string {
//
//}
